package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Demo3Dianxin {


    Connection connection;

    /**
     * 创建链接
     */
    @Before
    public void init() {
        //1、创建hbase链接
        //创建hadoop配置对象
        Configuration configuration = HBaseConfiguration.create();
        //指定链接参数
        //指定zk地址，在zk中保存了hbase元数据
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        //建立链接
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void putAll() throws Exception {

        //获取表对象
        Table dianxin = connection.getTable(TableName.valueOf("dianxin"));

        ArrayList<Put> puts = new ArrayList<>();

        //读取文件
        BufferedReader bufferedReader = new BufferedReader(new FileReader("../data/DIANXIN.csv"));
        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            String mdn = split[0];//手机号
            String time = split[1];//时间
            String lat = split[4];//经度
            String lon = split[5];//维度

            String location = lat + "," + lon;

            //构建put,以手机号作为rowkey,以数据发生的时间作为版本号
            Put put = new Put(Bytes.toBytes(mdn));

            //增加一个列
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("location"), Long.parseLong(time), Bytes.toBytes(location));

            puts.add(put);
        }

        //批量插入数据
        dianxin.put(puts);
    }


    @Test
    public void query() throws Exception {
        Table dianxin = connection.getTable(TableName.valueOf("dianxin"));

        Get get = new Get(Bytes.toBytes("47BE1E866CFC071DB19D5E1C056BE28AE24C16E7"));
        //指定查询的版本的数量
        get.setMaxVersions(10);
        //查询数据
        Result result = dianxin.get(get);

        //获取所以的单元格
        List<Cell> cells = result.listCells();

        for (Cell cell : cells) {
            //解析单元格

            //去除value
            byte[] bytes = CellUtil.cloneValue(cell);
            String location = Bytes.toString(bytes);

            //取出时间
            long timestamp = cell.getTimestamp();

            System.out.println(timestamp + "\t" + location);
        }

    }


}
